package com.ruoyi.kafka;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

/**
 * @description kafka 配置信息工具类
 * @author arvin
 * @date 2025/1/20 10:56
 */
@Configuration
public class KafkaConfiguration {

    @Bean(name = "singleListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<?, ?> singleListenerContainerFactory(
            ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        containerFactory.setConsumerFactory(consumerFactory);
        containerFactory.setConcurrency(2);
        containerFactory.setBatchListener(false);
        containerFactory.setAutoStartup(true);
        containerFactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return containerFactory;
    }

//    @Bean(name = "batchListenerContainerFactory")
//    public ConcurrentKafkaListenerContainerFactory<?, ?> batchListenerContainerFactory(
//            ConsumerFactory<Object, Object> consumerFactory) {
//        ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
//        containerFactory.setConsumerFactory(consumerFactory);
//        // 设置为批量监听
//        containerFactory.setBatchListener(true);
//        containerFactory.setAutoStartup(true);
//        containerFactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
//
//        // 使用 SeekToCurrentBatchErrorHandler
//        SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler();
//        // 多次重试，每次间隔1000ms
//        batchErrorHandler.setBackOff(new FixedBackOff(1000L, Integer.MAX_VALUE));
//        containerFactory.setBatchErrorHandler(batchErrorHandler);
//        return containerFactory;
//    }
}
